package com.asap.demo.function;

import com.asap.demo.ContextInfo;
import com.asap.interf.Action;
import com.asap.rule.StandardEvent;
import com.asap.rule.engine.InferenceEngine;
import com.asap.rule.engine.PatternMatcher;
import com.asap.rule.util.RuleReader;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.ListTypeInfo;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author wangbh
 * @Description: todo
 * @date 2021/11/15 9:55
 */
public class dealStreamProcessFunctionList extends KeyedBroadcastProcessFunction<String, StandardEvent, List<String>, StandardEvent> {
    private static final Logger logger = LoggerFactory.getLogger(dealStreamProcessFunctionList.class);

    private transient List<StandardEvent> listState;
    private transient Boolean runingFlagState;
    private transient InferenceEngine engineState;
    MapStateDescriptor<String, List<String>> ruleStateDescriptor = new MapStateDescriptor<>(ContextInfo.RULE_SBROAD_CAST_STATE
            , BasicTypeInfo.STRING_TYPE_INFO
            , new ListTypeInfo<>(String.class));
    InferenceEngine engine;

    /**
     * open方法只会执行一次
     * 可以在这实现初始化的功能
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ValueStateDescriptor<List<StandardEvent>> recentOperatorsDescriptor = new ValueStateDescriptor<List<StandardEvent>>(
                "recent-operator",
                TypeInformation.of(new TypeHint<List<StandardEvent>>() {
                }));

        ValueStateDescriptor<Boolean> runingFlagDescriptor = new ValueStateDescriptor<Boolean>(
                "runingFlag",
                Boolean.class);

        ValueStateDescriptor<InferenceEngine> engineDescriptor = new ValueStateDescriptor<InferenceEngine>(
                "runingFlag1",
                InferenceEngine.class);
        engineState = null;
        listState = new ArrayList<>();
        runingFlagState = false;

        logger.info("KeyedBroadcastProcessFunction open");
    }

    @Override
    public void processElement(StandardEvent standardEvent, ReadOnlyContext readOnlyContext, Collector<StandardEvent> collector) throws Exception {
        if (standardEvent == null) {
            return;
        }
        List<String> list = null;
        list = readOnlyContext.getBroadcastState(ruleStateDescriptor).get(ContextInfo.RULE_SBROAD_CAST_STATE);
        if (list == null) {
            logger.info("RulesBroadcastState is null..............");
            listState.add(standardEvent);
            return;
        }
        //第一次进来
        if (!runingFlagState) {
            logger.info("runingFlagState.value() == null");
            runingFlagState = true;
        }
        if (((runingFlagState && list.get(0).equals("1")) || list.get(0).equals("0"))) {
            logger.info("action update.....:" + list.size() + ":" + runingFlagState + ":" + list.get(0));
            String flag = list.get(0);
            list.remove(0);
            engineState = InferenceEngine.compile(RuleReader.parseRules(list));
            if (runingFlagState && flag.equals("1")) {
                runingFlagState = false;
            }
        }

        if (engineState != null) {
            List<StandardEvent> listTmp = listState;
            if (listTmp != null) {
                for (StandardEvent standardEventTmp : listTmp) {
                    logger.info("listState.....:" + standardEventTmp);
                    match(standardEventTmp, collector);
                }
                listState.clear();
            }
            match(standardEvent, collector);
        } else {
            logger.info("processElement engine is null.....:");

        }

    }

    private void match(StandardEvent standardEvent, Collector<StandardEvent> collector) throws IOException {
        PatternMatcher matcher = engineState.matcher(standardEvent);
        if (matcher.find()) {
            List<Action> actions = matcher.getActions();
            boolean flag =false;
            StringBuilder sb = new StringBuilder();
            for (Action action : actions) {
                if (standardEvent != null) {
                    if (collector != null){
                        sb.append(action.getRuleId()).append(":");
                        flag =true;
                    }
                    else{
                        logger.info("collector is null:");
                    }
                }
            }
            if(flag){
                String tmp = sb.toString();
                standardEvent.getAllFields().put("rule_id",sb.toString().substring(0,tmp.length() -1));
                collector.collect(standardEvent);
            }
        } else {
            logger.info("no matcher:" + standardEvent);
        }
    }

    @Override
    public void processBroadcastElement(List<String> strings, Context context, Collector<StandardEvent> collector) throws Exception {
        BroadcastState<String, List<String>> broadcastState = context.getBroadcastState(ruleStateDescriptor);
        logger.info("processBroadcastElement.....:" + strings.size());
        if (broadcastState.contains(ContextInfo.RULE_SBROAD_CAST_STATE)) {
            List<String> oldList = broadcastState.get(ContextInfo.RULE_SBROAD_CAST_STATE);
            logger.info("get State:" + oldList.size() + "  replaced with State:" + strings.size());
        } else {
            logger.info("do not find old State, put first counterState {}", strings.size());
        }
        broadcastState.put(ContextInfo.RULE_SBROAD_CAST_STATE, strings);
    }
}

